//package anli
//
//import org.apache.spark.rdd.RDD
//import org.apache.spark.util.AccumulatorV2
//import org.apache.spark.{SparkConf, SparkContext}
//
//import scala.collection.mutable
///**
// * 改进程序：在没有shuffle的情况下完成，即不要使用reduceByKey，使用累加器
// * */
///**程序优化改进方案1：
// *   1、actionRDD重复使用，直接把重复使用的东西放在Driver中，即actionRDD.cache()
// *
// *   2、cogroup过程中存在shuffle操作，有可能降低性能,可将cogroup改为union并集
// * 程序优化改进方案2（在1的基础上改进）：
// *   1、将数据转换结构： 点击的场合结构 (品类ID,(1,0,0)),
// *                    下单场合结构 (品类ID ,(0,1,0))
// *                    支付场合结构 (品类ID ,(0,1,0))
// *    2、将相同的品类ID的数据进行分组聚合：(品类ID,(点击数量，下单数量，支付数量))
// *    3、将统计结果根据降序进行排序，取前十
// *   */
//object HotCategoryTop10Analysis_Gai2 {
//  /**
//   * 网站用户行为：搜索，点击，下单，支付
//   * 数据字段：日期 用户id SessionID 页面ID 动作时间 搜索关键字
//   * 需求:1、读取原始日志数据
//   * 2、统计品类的点击数量：品类ID和点击次数
//   * 3、统计品类的下单数量：品类ID，下单数量
//   * 4、统计品类的支付数量：品类ID，支付数量
//   * 5、将品类进行排序，并取前10名
//   *    点击数量排序，下单数量排序，支付数量排序
//   *    元组排序：先比较第一个，再比较第二个，再比较后面的，依此类推
//   * 6、将结果采集到控制台打印出来
//   * */
//  def main(args: Array[String]): Unit = {
//    val sparkconf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("anli2")
//    val sc=new SparkContext(sparkconf)
//    // 需求:1、读取原始日志数据
//    val actionRDD=sc.textFile("data/user_visit_action.txt")
//    //优化：直接把重复使用的东西放在Driver中
////    actionRDD.cache()
//    //注册累加器
//    val acc=new HotCategoryAccumulator()
//    sc.register(acc,"hotCategory")
//
//     actionRDD.foreach(
//      action => {
//        val datas = action.split("_")
//        if (datas(6) != "-1") {
//          //点击场合
//          acc.add((datas(6),"click"))
//        } else if (datas(8) != "null") {
//          //下单场合
//          val ids = datas(8).split(",")
//          ids.foreach(
//              id=>{
//                 acc.add((id,"order"))
//          }
//          )
//        } else if (datas(10) != "null") {
//          //支付场景
//          val ids = datas(10).split(",")
//          ids.foreach(
//            id=>{
//              acc.add(id,"pay")
//            }
//          )
//        }
//      }
//    )
//    val accVal: mutable.Map[String, HotCategory] = acc.value
//    val categories: mutable.Iterable[HotCategory] = accVal.map(_._2)
//    val sort: List[HotCategory] = categories.toList.sortWith(
//      (left, right) => {
//        if (left.clickCnt > right.clickCnt) {
//          true
//        } else if (left.orderCnt == right.orderCnt) {
//          if (left.orderCnt > right.orderCnt) {
//            true
//          } else if (left.orderCnt == right.orderCnt) {
//            left.payCnt > right.payCnt
//          } else {
//            false
//          }
//        } else {
//          false
//        }
//      }
//    )
//    sort.take(10).foreach(println)
//    sc.stop()
//  }
//
//}
//case class HotCategory(cid:String,var clickCnt:Int, var orderCnt:Int,var payCnt:Int)
///**
// * 自定义累加器
// * 1、继承AccumulatorV2,定义泛型
// * 传入的参数定义：
// *    IN：（品类ID，行为类型）
// *    OUT：mutable.Map[String,HotCategory]
// *2、重写方法，有6个
// * */
//class HotCategoryAccumulator extends AccumulatorV2[(String,String),mutable.Map[String,HotCategory]]{
// private val hcMap=mutable.Map[String,HotCategory]()
//  override def isZero: Boolean = {
//    hcMap.isEmpty
//  }
//
//  override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
//
//  new HotCategoryAccumulator()
//  }
//
//  override def reset(): Unit = {
//  hcMap.clear()
//  }
//
//  override def add(v: (String, String)): Unit = {
//    val cid=v._1
//    val actionType=v._2
//    val category: HotCategory = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
//    if(actionType=="click"){
//      category.clickCnt+=1
//    }else if(actionType=="pay"){
//      category.payCnt+=1
//    }else if(actionType=="order"){
//      category.orderCnt+=1
//    }
//    hcMap.update(cid,category)
//  }
//
//  override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
//    val map1=this.hcMap
//    val map2=other.value
//    map2.foreach{
//      case(cid,hc)=>{
//        val category: HotCategory = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
//        category.clickCnt+=hc.clickCnt
//        category.orderCnt+=hc.orderCnt
//        category.payCnt+=hc.payCnt
//        map1.update(cid,category)
//      }
//    }
//  }
//
//  override def value: mutable.Map[String, HotCategory] = hcMap
//}
